package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo5SourceFunction {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //使用自定义source
    val myDS: DataStream[String] = env.addSource(new MySource)

    myDS.print()

    env.execute()

  }
}

/**
 * 自定义Source, 实现SourceFunction接口
 */
class MySource extends SourceFunction[String] {
  /**
   * run,在任务启动时，执行一次，用于连接外部系统读取数据，再将数据发送到下游
   *
   * @param sourceContext :  用于将数据发送到下游
   */
  override def run(sourceContext: SourceFunction.SourceContext[String]): Unit = {

    while (true) {
      //将数据发送到下游
      sourceContext.collect("java")
      Thread.sleep(1000)
    }
  }

  //任务在取消的时候执行一次，一般用于回收资源
  override def cancel(): Unit = {}
}
